package com.zyg.async;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class CompletableFutureDemo {
    static ExecutorService executorService = Executors.newFixedThreadPool(100);

    public static void main(String[] args) {
//        //1.runAsync
//        CompletableFuture.runAsync(() -> {
//            //子任务,无返回值
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println(Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }, executorService);
//        System.out.println("主任务main end");
//        executorService.shutdown();

        //2.supplyAsync
//        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
//            try {
//                //int i = 1 / 0;
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f1" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//        System.out.println("f1 main end ");
//        try {
//            System.out.println(completableFuture.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }finally {
//            executorService.shutdown();
//        }

        /**
         * 异步计算结果触发回调
         */
        //3.whenComplete 根据异步任务的结果，触发业务逻辑
//        completableFuture.whenComplete((Integer integer,Throwable throwable)->{
//            System.out.println("异步任务后续处理线程---》"+Thread.currentThread().getName());
//            System.out.println(integer);
//        });
//        executorService.shutdown();

        //4.whenCompleteAsync
//        completableFuture.whenCompleteAsync((Integer integer,Throwable throwable)->{
//            System.out.println("异步任务后续处理线程---》"+Thread.currentThread().getName());
//            System.out.println(integer);
//            //executorService.shutdown();
//        });


//        completableFuture.whenCompleteAsync((Integer integer, Throwable throwable) -> {
//            System.out.println("异步任务后续处理线程---》" + Thread.currentThread().getName());
//            System.out.println(integer);
//            executorService.shutdown();
//        }, executorService);

//        completableFuture.exceptionally((Throwable throwable) -> {
//            System.out.println("异步结果处理线程----》" + Thread.currentThread().getName());
//            System.out.println(throwable);
//            executorService.shutdown();
//            return null;
//        });

        /**
         * 多任务依赖执行 流式调用链
         */
        //1.applyAsync多任务执行
//        CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f2"+Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService).thenApplyAsync(value->{
//            System.out.println("当前异步任务执行的线程"+Thread.currentThread().getName());
//            return value*10;
//        },executorService);
//        System.out.println("f2 main end ");
//        try {
//            System.out.println(completableFuture2.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }finally {
//            executorService.shutdown();
//        }
        //2.handle() 后面的链式调用子任务需要返回结果
//        CompletableFuture<Integer> completableFuture3 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f3 "+Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService).handle((Integer integer,Throwable throwable)->{
//            System.out.println("f3 "+Thread.currentThread().getName());
//            if (throwable==null){
//                return integer*10;
//            }else {
//                System.out.println(throwable.getMessage());
//            }
//            return null;
//        });
//        System.out.println("f3 main end ");
//        try {
//            System.out.println(completableFuture3.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }finally {
//            executorService.shutdown();
//        }
        //3.thenAccept() 链式调用不返回结果，但前面的任务须有结果
//        CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("异步任务执行的线程" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService).thenAccept(value -> {
//            System.out.println("异步任务执行的线程"+Thread.currentThread().getName());
//            System.out.println("前置任务的处理结果" + value);
//            executorService.shutdown();
//        });

//        CompletableFuture<Void> completableFuture4 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f4 异步任务执行的线程" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService).thenAcceptAsync(value -> {
//            System.out.println("f4 异步任务执行的线程"+Thread.currentThread().getName());
//            System.out.println("前置任务的处理结果" + value);
//            executorService.shutdown();
//        },executorService);
//        System.out.println("f4 main end ");

        //4.thenRun() 不关心前面是否有返回值
//        CompletableFuture<Void> completableFuture5 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f5 异步任务执行的线程" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService).thenRunAsync(() -> {
//            System.out.println("thenRun() 执行了");
//            System.out.println("f5 异步任务执行的线程" + Thread.currentThread().getName());
//            executorService.shutdown();
//        },executorService);
//        System.out.println("f5 main end ");


        /**
         * 合并前面多个异步任务的结果 前面的任务必须全部触发才能进行后续的逻辑处理
         */
        //thenCombine 必须等前面的任务都执行完毕 且之后的合并操作必须带返回值
//        CompletableFuture<Integer> completableFuture6 = CompletableFuture.supplyAsync(() -> {
//            try {
//                //int i = 1 / 0;
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f6" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//        System.out.println("f6 main end ");
//
//        CompletableFuture<Integer> completableFuture7 = CompletableFuture.supplyAsync(() -> {
//            try {
//                //int i = 1 / 0;
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f7 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//        System.out.println("f7 main end ");
//
//        CompletableFuture<Integer> completableFuture8 = completableFuture6.thenCombineAsync(completableFuture7, (f6, f7) -> {
//            System.out.println("合并f8 " + Thread.currentThread().getName());
//            executorService.shutdown();
//            return f6 + f7;
//        }, executorService);
//        try {
//            System.out.println(completableFuture8.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }

        //thenAcceptBoth 必须等前面的任务都执行完毕 之后的合并操作不带返回值
//        CompletableFuture<Integer> completableFuture9 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f9" + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture10 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(2);
//                System.out.println("f10 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Void> completableFuture11 = completableFuture9.thenAcceptBothAsync(completableFuture10, (f9, f10) -> {
//            System.out.println("合并f11 " + Thread.currentThread().getName());
//            System.out.println(f9+f10);
//            executorService.shutdown();
//        }, executorService);
        //thenRunAfterBoth 不需要前面异步任务的执行结果，但需要等前面的任务执行完，后面的合并也不需要返回结果
//        CompletableFuture<Integer> completableFuture12 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f12 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture13 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f13 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Void> completableFuture14 = completableFuture12.runAfterBothAsync(completableFuture13, () -> {
//            System.out.println("合并f14 " + Thread.currentThread().getName());
//            executorService.shutdown();
//        }, executorService);


        /**
         * 多个任务只要完成一个即可触发后面的业务处理
         *
         */
//        CompletableFuture<Integer> completableFuture15 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(5);
//                System.out.println("f15 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture16 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f16 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture17 = completableFuture15.applyToEitherAsync(completableFuture16, (value) -> {
//            System.out.println("合并f17 " + Thread.currentThread().getName());
//            executorService.shutdown();
//            return value;
//        }, executorService);
//
//        try {
//            System.out.println(completableFuture17.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }

        //acceptEither对比上述的acceptBoth runAferEither对比runAfterBoth
//        CompletableFuture<Integer> completableFuture18 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(5);
//                System.out.println("f18 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture19 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(2);
//                System.out.println("f19 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Void> completableFuture20 = completableFuture18.acceptEitherAsync(completableFuture19, (value) -> {
//            System.out.println("合并f20 " + Thread.currentThread().getName());
//            System.out.println(value);
//            executorService.shutdown();
//        }, executorService);

//
//        CompletableFuture<Integer> completableFuture21 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(5);
//                System.out.println("f21 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture22 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f22 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Void> completableFuture23 = completableFuture21.runAfterEitherAsync(completableFuture22, () -> {
//            System.out.println("合并f23 " + Thread.currentThread().getName());
//            executorService.shutdown();
//        }, executorService);

        //多任务都完成后才执行后续操作
        //1.allOf() 底层采用了二叉树的方式当前所有任务执行完后才会继续往下执行
//        CompletableFuture<Integer> completableFuture24 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(3);
//                System.out.println("f24 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 123;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture25 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(5);
//                System.out.println("f25 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 234;
//        }, executorService);
//
//        CompletableFuture<Integer> completableFuture26 = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(4);
//                System.out.println("f26 " + Thread.currentThread().getName());
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            return 345;
//        }, executorService);
//
//        List<CompletableFuture<Integer>> completableFutures = new ArrayList<>();
//        completableFutures.add(completableFuture24);
//        completableFutures.add(completableFuture25);
//        completableFutures.add(completableFuture26);
//        CompletableFuture<Void> all = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[]{}));
//        all.thenRunAsync(() -> {
//            AtomicInteger atomicInteger = new AtomicInteger();
//            completableFutures.stream().forEach(
//                    completableFuture -> {
//                        try {
//                            int value = completableFuture.get();
//                            atomicInteger.updateAndGet(v->v+value);
//                            System.out.println(atomicInteger);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        } catch (ExecutionException e) {
//                            e.printStackTrace();
//                        }
//                    }
//            );
//            //System.out.println(atomicInteger);
//            executorService.shutdown();
//        }, executorService);

        //2.anyOf() 底层采用了二叉树的方式当前有一个任务执行完后就执行，获取最快的执行线程的结果
        CompletableFuture<Integer> completableFuture27 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println("f27 " + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 123;
        }, executorService);

        CompletableFuture<Integer> completableFuture28 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
                System.out.println("f28 " + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 234;
        }, executorService);

        CompletableFuture<Integer> completableFuture29 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(4);
                System.out.println("f29 " + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 345;
        }, executorService);

        List<CompletableFuture<Integer>> completableFutures = new ArrayList<>();
        completableFutures.add(completableFuture27);
        completableFutures.add(completableFuture28);
        completableFutures.add(completableFuture29);
        CompletableFuture<Object> any = CompletableFuture.anyOf(completableFutures.toArray(new CompletableFuture[]{}));
        any.thenRunAsync(() -> {
            try {
                System.out.println("最快的执行结果是 " + any.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            executorService.shutdown();
        }, executorService);
    }
}

